package com.gitee.ixtf.vertx.rabbitmq.verticle

import com.gitee.ixtf.vertx.rabbitmq.AMQ_TOPIC
import io.vertx.core.Future
import io.vertx.kotlin.core.json.jsonObjectOf
import io.vertx.kotlin.coroutines.toReceiveChannel
import io.vertx.rabbitmq.QueueOptions
import io.vertx.rabbitmq.RabbitMQConsumer
import io.vertx.rabbitmq.RabbitMQMessage
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.launch

abstract class RabbitMQConsumerVerticle : RabbitMQVerticle() {
  protected abstract val queue: String
  protected abstract val error: String
  protected lateinit var consumer: RabbitMQConsumer
  protected open val onSuccessAck = true

  protected abstract suspend fun handle(delivery: RabbitMQMessage)

  override fun prepareClient(): Future<Void> =
      super.prepareClient()
          .compose { client.queueDeclare(error, true, false, false) }
          .compose { client.queueDeclare(queue, true, false, false, config()) }
          .compose { client.queueBind(queue, AMQ_TOPIC, "$queue.*") }
          .compose { client.basicConsumer(queue, QueueOptions().setAutoAck(false)) }
          .compose {
            consumer = it
            launch { prepareConsumer() }
            Future.succeededFuture()
          }

  protected open suspend fun prepareConsumer() {
    consumer.toReceiveChannel(vertx).consumeEach { delivery ->
      launch {
        runCatching { handle(delivery) }
            .onSuccess { if (onSuccessAck) delivery.basicAck() }
            .onFailure { delivery.basicNack(it) }
      }
    }
  }

  protected open fun config() =
      jsonObjectOf(
          "x-dead-letter-exchange" to "",
          "x-dead-letter-routing-key" to error,
          // "x-max-priority" to 10
      )
}
